import selectors
import time
from collections import deque
import socket
import sys
from typing import List, Any, Coroutine


class Future:
    # """表示一个异步操作的未来结果"""
    #  future的理解： 1. 是个占位符， 2. 可以进行状态追踪， 3. 如果完成了可以得到结果
    def __init__(self):
        self._result = None      # 存储操作结果
        self._exception = None   # 存储异常对象
        self._callbacks = []     # 存储操作完成时的回调函数
        self._done = False       # 标记操作是否完成

    def result(self):
        """获取结果，如果未完成则抛出异常，如果操作以异常结束则抛出该异常"""
        if not self._done:
            raise RuntimeError("Future not done yet")
        if self._exception is not None:
            raise self._exception
        return self._result

    def exception(self):
        """获取异常对象，如果没有异常或未完成则返回None"""
        if not self._done:
            raise RuntimeError("Future not done yet")
        return self._exception

    def add_done_callback(self, fn):
        """添加完成时的回调函数"""
        if self._done:
            fn(self)  # 如果已完成，立即调用
        else:
            self._callbacks.append(fn)  # 否则添加到回调列表

    def set_result(self, result):
        """设置结果并触发所有回调"""
        if self._done:
            raise RuntimeError("Future already done")
        self._result = result
        self._done = True
        for callback in self._callbacks:
            callback(self)  # 执行所有回调函数

    def set_exception(self, exception):
        """设置异常并触发所有回调"""
        if self._done:
            raise RuntimeError("Future already done")
        self._exception = exception
        self._done = True
        for callback in self._callbacks:
            callback(self)  # 执行所有回调函数

    def done(self):
        """返回Future是否已完成"""
        return self._done

    def __await__(self):
        """使Future对象可await"""
        yield self  # 暂停协程执行，直到Future完成
        return self.result()  # 返回结果或抛出异常


class Task(Future):
    # """管理协程执行的包装器"""
    # 是个包装器， 管理协程执行
    def __init__(self, coro, loop):
        super().__init__()
        self.coro = coro  # 要执行的协程
        self.loop = loop  # 所属事件循环
        self._step()      # 开始执行协程

    def _step(self, future=None):
        """执行协程的下一步"""
        try:
            if future is None:
                # 首次执行，启动协程
                result = self.coro.send(None)
            else:
                # 继续执行，传入之前的结果
                result = self.coro.send(future.result())
        except StopIteration as exc:
            # 协程执行完成，设置最终结果
            self.set_result(exc.value if exc.value is not None else None)
            return

        if isinstance(result, Future):
            # 如果协程yield了一个Future，添加回调继续执行
            result.add_done_callback(self._step)


class EventLoop:
    """事件循环核心实现"""
    def __init__(self):
        # Windows平台特殊处理
        if sys.platform == 'win32':
            self._wsa_initialized = hasattr(socket, 'WSAStartup')
            if self._wsa_initialized:
                wsa_version = 0x202  # Version 2.2
                socket.WSAStartup(wsa_version)

        self._selector = selectors.DefaultSelector()  # IO多路复用选择器
        self._ready = deque()        # 准备就绪的回调队列
        self._stopping = False       # 停止标志
        self._scheduled = []         # 定时任务列表
        self._current_handles = set()  # 当前处理的文件描述符集合

    def call_soon(self, callback, *args):
        """立即调度回调函数"""
        self._ready.append((callback, args))

    def call_later(self, delay, callback, *args):
        """延迟调度回调函数"""
        t = time.time() + delay
        self._scheduled.append((t, callback, args))
        self._scheduled.sort(key=lambda x: x[0])  # 按时间排序

    def create_task(self, coro):
        """创建任务"""
        task = Task(coro, self)
        return task

    def wait_for(self, fut, timeout):
        """
        等待一个 Future 或协程完成，带超时功能
        :param fut: 可以是 Future 或协程对象
        :param timeout: 超时时间(秒)
        :return: Future 的结果
        :raises: TimeoutError 如果超时
        """
        # 如果输入是协程，转换为 Task
        if not isinstance(fut, Future):
            fut = self.create_task(fut)

        result_future = Future()  # 用于返回最终结果
        timeout_handle = None

        def on_timeout():
            if not result_future.done():
                result_future.set_exception(TimeoutError())
                # 取消原始任务(如果支持取消)
                if hasattr(fut, 'cancel'):
                    fut.cancel()

        def on_completion(f):
            # 取消超时回调
            if timeout_handle is not None:
                timeout_handle.cancel()

            if not result_future.done():
                if f.exception():
                    result_future.set_exception(f.exception())
                else:
                    result_future.set_result(f.result())

        # 设置超时回调
        if timeout is not None:
            timeout_handle = self.call_later(timeout, on_timeout)

        # 添加完成回调
        fut.add_done_callback(on_completion)

        return result_future

    async def safe_gather(self, *coros, timeout=None):
        gather_future = self.gather(*coros)
        if timeout is None:
            return await gather_future

        return await self.wait_for(gather_future, timeout)

    async def gather(self, *coros: List[Coroutine]) :
        """类似asyncio.gather的功能，等待多个协程完成并返回结果列表"""
        futures = [self.create_task(coro) for coro in coros]  # 创建子任务的 Future
        results = [None] * len(futures)
        remaining = len(futures)
        # 是一个包装了所有子任务结果的 Future 对象。用来跟踪所有任务的完成状态
        future = Future()

        def _done_callback(i, fut):
            nonlocal remaining   # nonlocal 声明非局部变量， 内涵数修改外函数变量
            try:
                results[i] = fut.result()
            except Exception as e:
                results[i] = e      # 捕获异常，存储异常对象

            remaining -= 1
            if remaining == 0:
                future.set_result(results)

        for i, fut in enumerate(futures):
            # 每个子任务完成后，将结果存入 results 列表。
            # 当所有子任务完成时，触发总 future 的完成：
            fut.add_done_callback(lambda f, idx=i: _done_callback(idx, f))

        # await 关键字的解包作用; await 的核心功能是 解包 Future 对象并获取其最终结果
        # 当 future 完成时，await future 会返回 future.result()。
        # 等价于： result = future.result()  # 获取 Future 的结果  ---->  return result # 返回结果
        #  ----------》》》》  返回解包后的结果是异步编程中的通用实践  《《《------------
        return await future  # 等待所有任务完成，并返回 future 的结果

    #    添加超时处理避免死锁问题  ----》 如下代码替换：return await future
    #     try:
    #         if timeout:
    #             # 添加超时逻辑
    #             timeout_fut = loop.call_later(timeout, lambda: future.set_exception(TimeoutError()))
    #             return await future
    #         else:
    #             return await future
    #     except TimeoutError:
    #         # 取消所有未完成的任务
    #         for fut in futures:
    #             if not fut.done():
    #                 fut.cancel()
    #         raise

    def run_forever(self):
        """运行事件循环直到停止"""
        while not self._stopping:
            if not self._ready and not self._scheduled:
                break
            self._run_once()

    def _run_once(self):
        """执行一次事件循环"""
        # 处理准备就绪的回调
        ntodo = len(self._ready)
        for _ in range(ntodo):
            callback, args = self._ready.popleft()
            callback(*args)

        # 处理到期的定时任务
        now = time.time()
        while self._scheduled and self._scheduled[0][0] <= now:
            t, callback, args = self._scheduled.pop(0)
            self.call_soon(callback, *args)

        # 计算selector超时时间
        timeout = None
        if self._scheduled:
            timeout = max(0, self._scheduled[0][0] - time.time())
        if self._ready:
            timeout = 0  # 有就绪任务，不阻塞

        # IO多路复用等待事件
        if not self._ready:
            try:
                events = self._selector.select(timeout)
                for key, mask in events:
                    callback = key.data
                    if key.fileobj in self._current_handles:
                        self.call_soon(callback, key.fileobj, mask)
            except OSError as e:
                if getattr(e, 'winerror', None) != 10022:  # 忽略Windows上的无效参数错误
                    raise

    def stop(self):
        """停止事件循环"""
        self._stopping = True

    # IO相关方法
    def add_reader(self, fd, callback):
        """注册读事件监听"""
        self._selector.register(fd, selectors.EVENT_READ, callback)
        self._current_handles.add(fd)

    def remove_reader(self, fd):
        """移除读事件监听"""
        try:
            self._selector.unregister(fd)
            self._current_handles.discard(fd)
        except KeyError:
            pass

    def add_writer(self, fd, callback):
        """注册写事件监听"""
        self._selector.register(fd, selectors.EVENT_WRITE, callback)
        self._current_handles.add(fd)

    def remove_writer(self, fd):
        """移除写事件监听"""
        try:
            self._selector.unregister(fd)
            self._current_handles.discard(fd)
        except KeyError:
            pass

    def run_until_complete(self, coro):
        """运行直到协程完成"""
        task = self.create_task(coro)
        self.run_forever()
        return task.result()

    def close(self):
        """清理资源"""
        self._selector.close()
        if hasattr(self, '_wsa_initialized') and self._wsa_initialized:
            socket.WSACleanup()


def sleep(delay):
    """异步睡眠函数"""
    future = Future()
    loop.call_later(delay, lambda: future.set_result(None))
    return future


async def example_task(name, seconds):
    """示例任务，模拟耗时操作"""
    print(f"任务 {name} 开始")
    await sleep(seconds)
    print(f"任务 {name} 在 {seconds} 秒后完成")
    return f"{name}-结果"


async def tcp_echo_client(message):
    """TCP客户端示例"""
    # 创建非阻塞socket
    reader = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    reader.setblocking(False)

    try:
        reader.connect(('localhost', 8888))
    except BlockingIOError:
        pass  # 非阻塞连接会立即返回

    # 等待连接建立(可写)
    future = Future()
    loop.add_writer(reader, lambda _: future.set_result(None))
    await future
    loop.remove_writer(reader)

    print(f"发送: {message!r}")
    reader.send(message.encode())

    # 等待数据可读
    future = Future()
    loop.add_reader(reader, lambda _: future.set_result(None))
    await future
    loop.remove_reader(reader)

    data = reader.recv(100)
    print(f"收到: {data.decode()!r}")
    reader.close()
    return data.decode()


async def main():
    """主协程"""
    # 演示gather功能
    print("--- 测试gather ---")
    start = time.time()
    # results = await loop.gather(
    #     example_task("A", 5),
    #     example_task("B", 2),
    #     example_task("c", 4),
    #     example_task("d", 1),
    #     example_task("e", 3)
    # )

    # 另一种写法：
    # task1 = loop.create_task(coro1())
    # task2 = loop.create_task(coro2())
    # results = await gather(task1, task2)  # 仍然返回结果列表，但 task1/task2 是 Future
    # 这时可以中途 使用 task1 这个 Future 对像取消任务

    # 避免死锁的写法
    results = await loop.safe_gather(
            example_task("A", 5),
            example_task("B", 2),
            example_task("c", 4),
            example_task("d", 1),
            example_task("e", 3), timeout=6
    )
    print(time.time() - start)

    print(type(results))

    print("所有任务完成:", results)

    # 演示网络IO功能
    # print("\n--- 测试网络IO ---")
    # try:
    #     result = await tcp_echo_client("Hello World")
    #     print("网络操作结果:", result)
    # except Exception as e:
    #     print(f"网络错误: {e}")

# 创建并运行事件循环
loop = EventLoop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()